package cn.texous.demo.dmqs.consumer;

import cn.texous.demo.dmqs.config.NsqConfig;
import cn.texous.demo.dmqs.handler.MessageHandler;
import com.github.brainlag.nsq.NSQConsumer;
import com.github.brainlag.nsq.NSQMessage;
import com.github.brainlag.nsq.callbacks.NSQMessageCallback;
import com.github.brainlag.nsq.lookup.DefaultNSQLookup;
import com.github.brainlag.nsq.lookup.NSQLookup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * insert description here
 *
 * @author Showa.L
 * @since 2019/8/6 14:53
 */
@Slf4j
//@Component
public class NsqConsumer {

    @Autowired
    private NsqConfig nsqConfig;
    @Autowired
    private MessageHandler nsqMessageHandler;

    //    @PostConstruct
    private void init() {
        nsqConsumer(nsqConfig.getTpinfos().get("test").get(0));
    }

    /**
     * nsq 消费者
     * @param topic topic
     */
    public void nsqConsumer(String topic) {
        NSQLookup lookup = new DefaultNSQLookup();
        lookup.addLookupAddress(nsqConfig.getHost(), nsqConfig.getConsumerPort());
        NSQConsumer consumer = new NSQConsumer(lookup, topic,
                nsqConfig.getConsumerChannel(), new MyNsqMessageCallback());
        Executor executor = Executors.newFixedThreadPool(4);
        consumer.setExecutor(executor);

        consumer.start();
    }

    private class MyNsqMessageCallback implements NSQMessageCallback {

        private AtomicInteger integer = new AtomicInteger();

        @Override
        public void message(NSQMessage message) {
            System.out.println(Thread.currentThread().getName());
            log.info("接收到的第 {} 条消息", integer.incrementAndGet());
            byte[] b = message.getMessage();
            String s = new String(b);
            System.out.println(s);
            nsqMessageHandler.handler(s);
            message.finished();
        }
    }

}
